package io.kiki.sba.registry.store.jdbc.repository.impl;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.Sets;
import io.kiki.sba.registry.cache.CacheCleaner;
import io.kiki.sba.registry.common.model.store.AppRevision;
import io.kiki.sba.registry.concurrent.CachedExecutor;
import io.kiki.sba.registry.store.api.config.DefaultCommonConfig;
import io.kiki.sba.registry.store.api.date.DateNowRepository;
import io.kiki.sba.registry.store.api.meta.RecoverConfig;
import io.kiki.sba.registry.store.api.repository.AppRevisionRepository;
import io.kiki.sba.registry.store.jdbc.constant.TableEnum;
import io.kiki.sba.registry.store.jdbc.convertor.AppRevisionDomainConvertor;
import io.kiki.sba.registry.store.jdbc.domain.AppRevisionDomain;
import io.kiki.sba.registry.store.jdbc.exception.RevisionNotExistException;
import io.kiki.sba.registry.store.jdbc.informer.BaseInformer;
import io.kiki.sba.registry.store.jdbc.mapper.AppRevisionMapper;
import io.kiki.sba.registry.util.NamedThreadFactory;
import io.kiki.sba.registry.util.StringFormatter;
import io.kiki.sba.registry.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import static io.kiki.sba.registry.store.jdbc.repository.impl.MetadataMetrics.Fetch.REVISION_CACHE_HIT_COUNTER;
import static io.kiki.sba.registry.store.jdbc.repository.impl.MetadataMetrics.Fetch.REVISION_CACHE_MISS_COUNTER;
import static io.kiki.sba.registry.store.jdbc.repository.impl.MetadataMetrics.Register.REVISION_REGISTER_COUNTER;


public class AppRevisionJdbcRepository implements AppRevisionRepository, RecoverConfig {

    private static final Logger logger = LoggerFactory.getLogger(AppRevisionJdbcRepository.class);

    /**
     * map: <revision, AppRevision>
     */
    private final LoadingCache<String, AppRevision> registry;

    private final Cache<String, Boolean> localRevisions = CacheBuilder.newBuilder().expireAfterWrite(2, TimeUnit.MINUTES).build();

    private final CachedExecutor<String, Boolean> cachedExecutor = new CachedExecutor<>(1000 * 10);

    private final ScheduledExecutorService revisionDigestService = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("RevisionDigest"));
    private final Informer informer = new Informer();
    @Autowired
    private AppRevisionMapper appRevisionMapper;
    @Resource
    private InterfaceAppsJdbcRepository interfaceAppsJdbcRepository;
    @Autowired
    private DateNowRepository dateNowRepository;
    @Autowired
    private DefaultCommonConfig defaultCommonConfig;
    private Set<String> dataCenters = Sets.newConcurrentHashSet();

    public AppRevisionJdbcRepository() {
        this.registry = CacheBuilder.newBuilder().maximumSize(10000L).expireAfterAccess(60, TimeUnit.MINUTES).build(new CacheLoader<String, AppRevision>() {
            @Override
            public AppRevision load(String revision) throws InterruptedException {
                StringUtil.checkNotEmpty(dataCenters, "dataCenters");

                REVISION_CACHE_MISS_COUNTER.inc();
                List<AppRevisionDomain> revisionDomains = appRevisionMapper.queryRevision(dataCenters, revision);
                if (CollectionUtils.isEmpty(revisionDomains)) {
                    throw new RevisionNotExistException(revision);
                }
                for (AppRevisionDomain revisionDomain : revisionDomains) {
                    if (!revisionDomain.isDeleted()) {
                        return AppRevisionDomainConvertor.convert2Revision(revisionDomain);
                    }
                }
                throw new RevisionNotExistException(revision);
            }
        });
        CacheCleaner.autoClean(localRevisions, 1000 * 60 * 10);
    }

    @PostConstruct
    public void init() {

        revisionDigestService.scheduleAtFixedRate(() -> {
            try {
                logger.info("informer revision size: {}", informer.getContainer().size());
            } catch (Throwable t) {
                logger.error("informer revision size digest error", (Throwable) t);
            }
        }, 60, 60, TimeUnit.SECONDS);
    }

    @Override
    public void register(AppRevision appRevision) throws Exception {
        if (appRevision == null) {
            throw new RuntimeException("jdbc register app revision error, appRevision is null.");
        }
        interfaceAppsJdbcRepository.register(appRevision.getAppName(), appRevision.getInterfaceMap().keySet());

        localRevisions.put(appRevision.getRevision(), true);
        if (informer.getContainer().containsRevisionId(appRevision.getRevision())) {
            return;
        }
        AppRevisionDomain domain = AppRevisionDomainConvertor.convert2Domain(defaultCommonConfig.getDefaultClusterId(), appRevision);
        // new revision, save into database
        REVISION_REGISTER_COUNTER.inc();
        refreshEntryToStorage(domain);
    }

    /**
     * check if revisionId exist
     *
     * @param revisionId revisionId
     * @return boolean
     */
    @Override
    public boolean exist(String revisionId) {
        return informer.getContainer().containsRevisionId(revisionId);
    }

    @Override
    public AppRevision queryRevision(String revision) {
        AppRevision appRevision;
        appRevision = registry.getIfPresent(revision);
        if (appRevision != null) {
            REVISION_CACHE_HIT_COUNTER.inc();
            return appRevision;
        }
        try {
            return registry.get(revision);
        } catch (ExecutionException e) {
            logger.error("jdbc query revision error, revision: {}", revision, e);
            throw new RuntimeException("jdbc refresh revision failed", e);
        } catch (Throwable t) {
            if (t.getCause() instanceof RevisionNotExistException) {
                logger.info("jdbc query revision failed, revision: {} not exist in db", revision, t);
            }
        }
        return null;
    }

    @Override
    public boolean heartbeat(String revision) {
        localRevisions.put(revision, true);
        return informer.getContainer().containsRevisionId(revision);
    }

    @Override
    public boolean heartbeatDB(String revision) {
        int effect = appRevisionMapper.heartbeat(defaultCommonConfig.getClusterId(tableName()), revision);
        if (effect == 0) {
            logger.error("revision: {} heartbeat fail.", revision);
        }
        return effect > 0;
    }


    void cleanCache() {
        registry.invalidateAll();
        cachedExecutor.clean();
    }

    @Override
    public Map<String, Integer> countByApp() {
        Map<String, Integer> counts = new HashMap<>();
        informer.getContainer().foreach((String revision, String appname) -> {
            counts.put(appname, counts.getOrDefault(appname, 0) + 1);
        });
        return counts;
    }

    @Override
    public Set<String> allRevisionIds() {
        return informer.getContainer().allRevisionIds();
    }

    protected void refreshEntryToStorage(AppRevisionDomain entry) {
        try {
            cachedExecutor.execute(entry.getRevision(), () -> {
                if (appRevisionMapper.heartbeat(entry.getDataCenter(), entry.getRevision()) == 0) {
                    appRevisionMapper.replace(entry);
                }
                logger.info("insert revision {}, succeed", entry.getRevision());
                return true;
            });
        } catch (Exception e) {
            logger.error("refresh to db failed: ", e);
            throw new RuntimeException(StringFormatter.format("refresh to db failed: {}", e.getMessage()));
        }
    }

    public Collection<String> availableRevisions() {
        return localRevisions.asMap().keySet();
    }

    @Override
    public List<AppRevision> listFromStorage(long start, int limit) {
        List<AppRevisionDomain> domains = appRevisionMapper.listRevisions(defaultCommonConfig.getDefaultClusterId(), start, limit);
        return AppRevisionDomainConvertor.convert2Revisions(domains);
    }

    @Override
    public void startSynced() {
        StringUtil.checkNotEmpty(dataCenters, "dataCenters");

        // after set datacenters
        informer.setEnabled(true);
        informer.start();
    }

    @Override
    public void waitSynced() {
        informer.waitSynced();
    }

    @Override
    public List<AppRevision> getExpired(Date beforeTime, int limit) {
        List<AppRevisionDomain> expired = appRevisionMapper.getExpired(defaultCommonConfig.getDefaultClusterId(), beforeTime, limit);
        return AppRevisionDomainConvertor.convert2Revisions(expired);
    }

    @Override
    public void replace(AppRevision appRevision) {
        appRevisionMapper.replace(AppRevisionDomainConvertor.convert2Domain(defaultCommonConfig.getDefaultClusterId(), appRevision));
    }

    @Override
    public int cleanDeleted(Date beforeTime, int limit) {
        return appRevisionMapper.cleanDeleted(defaultCommonConfig.getDefaultClusterId(), beforeTime, limit);
    }

    @Override
    public String tableName() {
        return TableEnum.app_revision.getTableName();
    }

    @Override
    public Set<String> dataCenters() {
        return new HashSet<>(dataCenters);
    }

    @Override
    public synchronized void setDataCenters(Set<String> dataCenters) {
        if (!this.dataCenters.equals(dataCenters)) {
            logger.info("dataCenters change from {} to {}", this.dataCenters, dataCenters);
            this.dataCenters = dataCenters;
        }
    }

    class Informer extends BaseInformer<AppRevisionDomain, AppRevisionContainer> {

        public Informer() {
            super("AppRevision");
        }

        @Override
        protected AppRevisionContainer containerFactory() {
            return new AppRevisionContainer();
        }

        @Override
        protected List<AppRevisionDomain> listFromStorage(long start, int limit) {
            return appRevisionMapper.listRevisions(defaultCommonConfig.getDefaultClusterId(), start, limit);
        }

        @Override
        protected Date getNow() {
            return dateNowRepository.getNow();
        }
    }
}
